package org.opens.lambda.future;

import com.google.common.collect.Maps;
import org.junit.Test;

import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 功能: 测试`线程池`、`访问数据库`、`CompletableFuture`如何应用到一起.
 *
 * @author zhangyuming@e6yun.com
 * @date 2022/5/27 16:22
 */
public class CompletableFutureAndFixedThreadPool {

    private ExecutorService executorService = Executors.newFixedThreadPool(5);

    /**
     * 功能: 测试CompletableFuture搭配线程池同时运行多个任务.
     * 输出:
     * ```
     * {count1=1, count2=2}
     * ```
     */
    @Test
    public void testRunAsync() {
        Map<String, Object> map = Maps.newHashMap();

        CompletableFuture<Void> runAsync1 = CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            map.put("count1", 1);
        }, executorService);

        CompletableFuture<Void> runAsync2 = CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            map.put("count2", 2);
        }, executorService);

        // 使主线程等待这两个异步任务的完成
        CompletableFuture.allOf(runAsync1, runAsync2).join();

        System.out.println(map);
    }

    /**
     * 功能: 测试CompletableFuture.anyOf等待任意一个任务完成立马结束.
     * 输出:
     * ```
     * {count1=1}
     * ```
     * 结论:
     * - 可以发现, 主线程并没有继续等待runAsync2任务的结束.
     */
    @Test
    public void testRunAsyncAndJoinAny() {
        Map<String, Object> map = Maps.newHashMap();

        CompletableFuture<Void> runAsync1 = CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            map.put("count1", 1);
        }, executorService);

        CompletableFuture<Void> runAsync2 = CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            map.put("count2", 2);
        }, executorService);

        // 使主线程等待这两个异步任务的完成
        CompletableFuture.anyOf(runAsync1, runAsync2).join();

        System.out.println(map);
    }

    /**
     * 功能: 测试CompletableFuture.supplyAsync执行异步任务从数据库中取值, 并使用join方法返回.
     * 输出:
     * ```
     * log: 获取key2 done!{key2=key2}
     * log: 获取key1 done!{key1=key1}
     * {key1=key1}
     * {key2=key2}
     * ```
     */
    @Test
    public void testGetDict() {
        CompletableFuture<Map<String, Object>> value1 = CompletableFuture.supplyAsync(() -> {
            return findDictByKey("key1");
        }).whenComplete((value, throwable) -> {
            Optional.ofNullable(value).ifPresent(v -> log("获取key1 done!" + v));
            Optional.ofNullable(throwable).ifPresent(t -> log("获取key1 发生异常: " + throwable.getMessage()));
        });

        CompletableFuture<Map<String, Object>> value2 = CompletableFuture.supplyAsync(() -> {
            return findDictByKey("key2");
        }).whenComplete((value, throwable) -> {
            Optional.ofNullable(value).ifPresent(v -> log("获取key2 done!" + v));
            Optional.ofNullable(throwable).ifPresent(t -> log("获取key1 发生异常: " + throwable.getMessage()));
        });;

        Map<String, Object> result1 = null;
        Map<String, Object> result2 = null;
        try {
            // java.util.concurrent.CompletableFuture.supplyAsync(java.util.function.Supplier<U>)方法传入的方法发生异常后在调用join方法时会抛出此异常.
            result1 = value1.join();
        } catch (Exception e) {
            System.out.println("发生异常1" + e.getMessage());
        }

        try {
            result2 = value2.join();
        } catch (Exception e) {
            System.out.println("发生异常2" + e.getMessage());
        }

        System.out.println("res1: " + result1);
        System.out.println("res2: " + result2);
    }

    @Test
    public void withAtomic() throws InterruptedException {
        //使用一个原子类来标记过程结束.
        AtomicBoolean finish = new AtomicBoolean(false);

        CompletableFuture<Map<String, Object>> completableFuture = CompletableFuture.supplyAsync(() -> {
            return findDictByKey("key1");
        }).whenComplete((v, t) -> {
            //如果存在结果就打印结果, 也就是说, 这里写异步任务执行完毕后的逻辑代码
            Optional.ofNullable(v).ifPresent(System.out::println);
            //如果存在异常则打印异常
            Optional.ofNullable(t).ifPresent(Throwable::printStackTrace);
            finish.set(true);
        });

        System.out.println("主线程输出1.");
        while (!finish.get()) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("主线程输出2.");
        Thread.currentThread().join();
    }

    /**
     * 功能: 模拟的读取数据库
     * @param key   读取的键
     * @return  模拟的数据库数据
     */
    public static Map<String, Object> findDictByKey(String key) {
        Map<String, Object> result = Maps.newHashMap();
        result.put(key, key);
        try {
            Thread.sleep(1000);
            //int a = 3232 / 0;
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return result;
    }

    public static void log(String message) {
        System.out.println("log: " + message);
    }

}
